package com.monster.datasourcejob.tablebuildingsql.generator.impl;

import com.monster.datasourcejob.model.Field;
import com.monster.datasourcejob.tablebuildingsql.generator.TableBuildingSqlGenerator;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.stream.Collectors;

/**
 * description:
 * hive建表语句生成器
 * @author Yuanml on 2021/6/28 18:05
 */
public class HiveTableBuildingSqlGenerator implements TableBuildingSqlGenerator {
    /**
     * 数据库名
     */
    private String dbName;
    /**
     * 表名
     */
    private String tableName;
    /**
     * 表字段
     */
    private List<Field> fields = null;
    /**
     * 分区字段
     */
    private List<Field> partitionField = null;
    /**
     * 行分隔符
     */
    private String terminatedField;
    /**
     * 存储格式
     */
    private String storedType;
    /**
     * 表注释
     */
    private String tableComment;

    private HiveTableBuildingSqlGenerator() {
    }

    @Override
    public String sqlForCreateTable() {
        return genSqlForCreateTable();
    }

    private String genSqlForCreateTable() {
        StringBuilder sqlBuilder = new StringBuilder();
        String tableName = this.tableName;
        if (StringUtils.isNotBlank(this.dbName)) {
            tableName = this.dbName + "." + tableName;
        }
        sqlBuilder.append("CREATE TABLE ").append("`").append(tableName).append("` ");
        // 1、处理表字段
        sqlBuilder.append(handleFields(fields));
        // 2、处理表注释
        if (StringUtils.isNotBlank(tableComment)) {
            sqlBuilder.append(" COMMENT '").append(tableComment).append("' ");
        }
        // 3、处理分区字段
        if (partitionField != null && partitionField.size() > 0) {
            sqlBuilder.append("PARTITIONED BY ").append(handleFields(partitionField)).append(" ");
        }
        // 4、处理分隔符
        sqlBuilder.append("row format delimited fields terminated by '").append(terminatedField).append("' ");
        sqlBuilder.append(";");
        return sqlBuilder.toString();
    }

    /**
     * description:
     * 将list集合中的fields转化为字符串："(`field1` type1 COMMENT 'comment1',`field2` type2 COMMENT 'comment2')"
     * @param fields    字段list
     * @return          返回字段list转化来的如下格式字符串："(`field1` type1 COMMENT 'comment1',`field2` type2 COMMENT 'comment2')"
     * @author Yuanml
     * @date 2021/6/29 11:18
     **/
    private String handleFields(List<Field> fields) {
        if (fields == null || fields.size() < 1) {
            throw new IllegalArgumentException("fields is required");
        }
        List<String> fieldStrList = fields.stream().map(field -> {
            checkField(field);
            StringBuilder fsb = new StringBuilder();
            fsb.append("`").append(field.getName()).append("` ");
            fsb.append(field.getType()).append(" ");
            if (StringUtils.isNotBlank(field.getComment())) {
                fsb.append("COMMENT '").append(field.getComment()).append("'");
            }
            return fsb.toString();
        }).collect(Collectors.toList());
        StringBuilder fsb = new StringBuilder("(");
        for (int i = 0; i < (fieldStrList.size() - 1); i++) {
            fsb.append(fieldStrList.get(i)).append(",");
        }
        fsb.append(fieldStrList.get(fieldStrList.size() - 1)).append(")");
        return fsb.toString();
    }

    private void checkField(Field field) {
        if (StringUtils.isBlank(field.getName())) {
            throw new IllegalArgumentException("name of field is required");
        }
        if (StringUtils.isBlank(field.getType())) {
            throw new IllegalArgumentException(String.format("type of field %s is required\n", field.getName()));
        }
    }


    /**
     * description:
     * hive建表语句生成器构造器
     * @author Yuanml
     * @date 2021/6/29 10:08
     **/
    public static class HiveTableSqlCreatorBuilder {
        private final HiveTableBuildingSqlGenerator hiveTableSqlCreator;
        public HiveTableSqlCreatorBuilder() {
            this.hiveTableSqlCreator = new HiveTableBuildingSqlGenerator();
        }
        public HiveTableSqlCreatorBuilder dbName(String dbName) {
            this.hiveTableSqlCreator.dbName = dbName;
            return this;
        }
        public HiveTableSqlCreatorBuilder tableName(String tableName) {
            this.hiveTableSqlCreator.tableName = tableName;
            return this;
        }
        public HiveTableSqlCreatorBuilder fields(List<Field> fields) {
            this.hiveTableSqlCreator.fields = fields;
            return this;
        }
        public HiveTableSqlCreatorBuilder partitionFields(List<Field> partitionField) {
            this.hiveTableSqlCreator.partitionField = partitionField;
            return this;
        }

        public HiveTableSqlCreatorBuilder terminatedField(String terminatedField) {
            this.hiveTableSqlCreator.terminatedField = terminatedField;
            return this;
        }

        public HiveTableSqlCreatorBuilder storedType(StorageType storedType) {
            this.hiveTableSqlCreator.storedType = storedType.typeName;
            return this;
        }

        public HiveTableSqlCreatorBuilder comment(String comment) {
            this.hiveTableSqlCreator.tableComment = comment;
            return this;
        }

        public HiveTableBuildingSqlGenerator build() {
            checkParams();
            return hiveTableSqlCreator;
        }

        private void checkParams() {
            if (StringUtils.isBlank(hiveTableSqlCreator.tableName)) {
                throw new IllegalStateException("name of table is required");
            }
            if (hiveTableSqlCreator.fields == null || hiveTableSqlCreator.fields.size() < 1) {
                throw new IllegalStateException("fields of table is required");
            }
            if (StringUtils.isBlank(hiveTableSqlCreator.terminatedField)) {
                throw new IllegalStateException("terminated field of table is required");
            }
        }
    }

    public enum StorageType {
        /**
         * TEXT_FILE
         */
        TEXT_FILE("TEXTFILE"),
        /**
         * ORC_FILE
         */
        ORC_FILE("ORCFILE"),
        /**
         * SEQUENCE_FILE
         */
        SEQUENCE_FILE("SEQUENCEFILE"),
        /**
         * RC_FILE
         */
        RC_FILE("RCFILE"),
        /**
         * PARQUET
         */
        PARQUET("PARQUET"),
        ;
        String typeName;

        StorageType(String typeName) {
            this.typeName = typeName;
        }
    }


}
